package kq

import (
	"goadmin-huasen-portal/util"

	"github.com/Shopify/sarama"
	"github.com/spf13/viper"
)

type Kafka struct {
	Topic    string
	Producer *sarama.SyncProducer
}

type HttpRequestLogEs struct {
	Type      string            `json:"type"`
	Url       string            `json:"url"`
	Header    map[string]string `json:"header"`
	Body      string            `json:"body"`
	Proxy     bool              `json:"proxy"`
	Response  string            `json:"response"`
	Error     string            `bson:"error,omitempty"`
	Date      string            `json:"date"`
	Time      string            `json:"time"`
	Timestamp string            `json:"timestamp"`
}

const (
	//HttpLogs http请求日志下发
	HttpLogs = "live_streaming_http_logs"
)

func NewKafka(topic string) Kafka {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 3
	config.Producer.Return.Successes = true

	producer, err := sarama.NewSyncProducer(viper.GetStringSlice("KqConf.Brokers"), config)
	if err != nil {
		util.Log().Error(err.Error())
	}

	return Kafka{
		Topic:    topic,
		Producer: &producer,
	}
}

func (m Kafka) SendLogs(topic string, message string) error {
	msg := &sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.StringEncoder(message),
	}

	p := *m.Producer
	partition, offset, err := p.SendMessage(msg)
	util.Log().Info("Send message to kafka topic %d:%d", partition, offset)
	if err != nil {
		util.Log().Error(message)
	}
	return err
}
